package com.service.data.log4j.appender.rocketmq

import com.service.data.commons.utils.Preconditions
import org.apache.log4j.AppenderSkeleton
import org.apache.log4j.spi.LoggingEvent
import org.apache.rocketmq.client.producer.{DefaultMQProducer, SendCallback, SendResult}
import org.apache.rocketmq.common.message.Message

import scala.beans.BeanProperty

/**
  * @author 伍鲜
  *
  *         Log4j Appender to RocketMQ
  */
class RocketMQAppender extends AppenderSkeleton {
  // 必填项
  @BeanProperty var namesrvAddr: String = _
  @BeanProperty var sendMessageMode: String = _

  // 建议填
  @BeanProperty var producerGroup: String = _
  @BeanProperty var topic: String = _

  // 可选项
  @BeanProperty var tags: String = ""
  @BeanProperty var keys: String = ""
  @BeanProperty var flag: Int = 0
  @BeanProperty var waitStoreMsgOK: Boolean = true
  @BeanProperty var retryTimesWhenSendAsyncFailed: Int = 2

  private var producer: DefaultMQProducer = _

  def init(): Unit = {
    namesrvAddr = Preconditions.checkNotNull(namesrvAddr, "Rocket MQ namesrvAddr must be specified. For example: log4j.appender.rocket.namesrvAddr=127.0.0.1:9876")
    sendMessageMode = Preconditions.checkNotNull(sendMessageMode, "Rocket MQ sendMessageMode must be specified. For example: log4j.appender.rocket.sendMessageMode=sync").toLowerCase
    producerGroup = Preconditions.getOrElse(producerGroup, "DEFAULT_PRODUCER_GROUP")
    topic = Preconditions.getOrElse(topic, "DEFAULT_TOPIC")

    producer = new DefaultMQProducer(producerGroup)
    producer.setNamesrvAddr(namesrvAddr)
    producer.start()

    if (sendMessageMode == "async") {
      producer.setRetryTimesWhenSendAsyncFailed(retryTimesWhenSendAsyncFailed)
    }
  }

  /**
    * 日志追加
    *
    * @param loggingEvent
    */
  override def append(loggingEvent: LoggingEvent): Unit = {
    if (producer == null) {
      RocketMQAppender.synchronized({
        if (producer == null) {
          init()
        }
      })
    }
    try {
      val msg = new Message(topic, tags, keys, flag, getLayout.format(loggingEvent).getBytes(), waitStoreMsgOK)
      sendMessageMode match {
        case "sync" =>
          // Send message in synchronous mode. This method returns only when the sending procedure totally completes.
          producer.send(msg)
        case "async" =>
          // Send Messages Asynchronously
          producer.send(msg, new SendCallback() {
            override def onException(e: Throwable): Unit = ???

            override def onSuccess(sendResult: SendResult): Unit = ???
          })
        case "oneway" =>
          // Send Messages in One-way Mode
          producer.sendOneway(msg)
        case _ =>
          throw new Exception("The value of the parameter sendMessageMode is set incorrectly.")
      }
    } catch {
      case _: Exception =>
    }
  }

  override def close(): Unit = {
    producer.shutdown()
  }

  override def requiresLayout(): Boolean = true
}

object RocketMQAppender {

}